package teamPL

import akka.actor.ActorSystem
import akka.event.{Logging, LoggingAdapter}
import akka.http.scaladsl.Http
import akka.stream.{ActorMaterializer, Materializer}
import com.typesafe.config.{Config, ConfigFactory}
import geotrellis.raster.{MultibandTile, Tile}
import geotrellis.spark.io.{AttributeStore, SpatialKeyFormat, ValueReader}
import geotrellis.spark.io.file.FileValueReader
import geotrellis.spark.{LayerId, SpatialKey}
import org.apache.spark.{SparkConf, SparkContext}
import teamPL.`trait`.ServiceSpark

import scala.concurrent.ExecutionContextExecutor

/**
 * TMS发布主程序
 * 时间：2022年9月28日20:07:17
 */
object MainSpark extends App with ServiceSpark {
  private val config: Config = ConfigFactory.load()

  //   监听接口
  private val port: Int = config.getInt("server.port")
  private val serverFilePath: String = config.getString("server.file.path")
  private val interface: String = config.getString("server.interface")

  //  spark Master
  private val master: String = config.getString("spark.master")
  //  spark 应用名称
  private val appName: String = config.getString("spark.appName")

  val sparkConf: SparkConf = new SparkConf()
    .setAppName(appName)
    .setMaster(master)
    .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .set("spark.kryo.registrator", "geotrellis.spark.io.kryo.KryoRegistrator")

  implicit val sparkContext: SparkContext = new SparkContext(sparkConf)

  val fileValueReader: FileValueReader = FileValueReader(serverFilePath)

  def reader(layerId: LayerId) = fileValueReader.reader[SpatialKey, MultibandTile](layerId)

  // val attributeStore: AttributeStore = AttributeStore(serverFilePath)
  // val valueReader: ValueReader[LayerId] = ValueReader(attributeStore,serverFilePath)
  // def reader(layerId: LayerId) = valueReader.reader[SpatialKey, MultibandTile](layerId)

  override implicit val system: ActorSystem = ActorSystem("tutorial-system")

  override implicit def executor: ExecutionContextExecutor = system.dispatcher

  override implicit val materializer: Materializer = ActorMaterializer()
  override val logger: LoggingAdapter = Logging(system, getClass)

  Http().bindAndHandle(root, interface, port)
}
